package com.ahcloud.edas.gatewaylog.server.kafka;

import com.ahcloud.edas.common.domain.common.LZ4CompressData;
import com.ahcloud.edas.common.kafka.AbstractConsumer;
import com.ahcloud.edas.common.kafka.client.KafkaMessageClient;
import com.ahcloud.edas.common.kafka.listener.MessageListener;
import com.ahcloud.edas.common.util.JsonUtils;
import com.ahcloud.edas.gatewaylog.biz.domain.log.dto.GatewayRequestLogDTO;
import com.ahcloud.edas.gatewaylog.biz.infrastructure.log.LogCollector;
import lombok.extern.slf4j.Slf4j;
import net.jpountz.lz4.LZ4Factory;
import net.jpountz.lz4.LZ4FastDecompressor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Objects;

/**
 * @program: ahcloud-operatelog
 * @description:
 * @author: YuKai Fan
 * @create: 2024/5/26 09:06
 **/
@Slf4j
@Component
public class GatewayAccessLogKafkaConsumer extends AbstractConsumer<String, String> {
    @Resource
    private LogCollector logCollector;

    private static final LZ4FastDecompressor decompressor = LZ4Factory.fastestInstance().fastDecompressor();

    @Autowired
    public GatewayAccessLogKafkaConsumer(KafkaMessageClient client) {
        super("gateway-logging-group", client);
    }

    @Override
    protected MessageListener<String, String> getMessageListenerInner() {
        return (consumer, record) -> {
            for (ConsumerRecord<String, String> result : record) {
                try {
                    LZ4CompressData lz4CompressData = JsonUtils.stringToBean(result.value(), LZ4CompressData.class);
                    if (Objects.isNull(lz4CompressData)) {
                        return;
                    }
                    byte[] decompress = decompressor.decompress(lz4CompressData.getCompressedData(), lz4CompressData.getLength());
                    logCollector.collect(JsonUtils.byteToBean(decompress, GatewayRequestLogDTO.class));
                } catch (Exception e) {
                    return;
                }
            }
        };
    }

    @Override
    protected String getTopic() {
        return "gateway-access-logging";
    }

    @Override
    public String getServiceName() {
        return this.getClass().getSimpleName();
    }
}
